//package com.xj.driver.mqtt.subscriber;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONArray;
//import com.alibaba.fastjson.JSONObject;
//import com.xj.common.bean.constant.Common;
//import com.xj.common.sdk.bean.driver.DriverContext;
//import com.xj.driver.mqtt.service.UpBizService;
//import lombok.extern.slf4j.Slf4j;
//import org.springframework.integration.annotation.ServiceActivator;
//import org.springframework.messaging.Message;
//import org.springframework.messaging.MessageHandler;
//import org.springframework.messaging.MessageHeaders;
//import org.springframework.messaging.MessagingException;
//import org.springframework.scheduling.annotation.Scheduled;
//import org.springframework.stereotype.Component;
//
//import javax.annotation.Resource;
//import java.util.concurrent.*;
//import java.util.concurrent.atomic.AtomicBoolean;
//
///**
// * @ClassName : MqttComManagerHandler
// * @Description : 设备接入
// * @Author : sunzhenzhen
// * @Date: 2024-05-28 15:37
// */
//@Slf4j
//@Component
//public class MqttComManagerHandler1 implements MessageHandler {
//
//
//
//    @Resource
//    private DriverContext driverContext;
//
//    @Resource
//    private UpBizService upBizService;
//
//
//
//    private final String primaryStationNo = "A"; //主用设备
//    private final String secondaryStationNo = "B"; //备用设备
//    // 设备状态跟踪
//    private final ConcurrentHashMap<String, Long> lastHeartbeatTimes = new ConcurrentHashMap<String, Long>() {{
//        put(primaryStationNo, 0L);   // 初始化主设备心跳时间
//        put(secondaryStationNo, 0L); // 初始化备设备心跳时间
//    }};
//    // 主备设备状态管理
//    private volatile  String activeStationNo = primaryStationNo; // 当前活跃设备ID
//
//    private final AtomicBoolean isSwitching = new AtomicBoolean(false);
//
//    // 配置参数（可通过@Value注入）
//    private static final long HEARTBEAT_TIMEOUT_MS = 400000; // 400秒超时
//    private static final long SWITCH_COOLDOWN_MS = 60000; // 60秒切换冷却时间
//    private volatile long lastSwitchTime = 0;
//
//    // 1. 创建消息缓冲队列
//    private final BlockingQueue<Message<?>> messageBuffer = new LinkedBlockingQueue<>(10000);
//    // 使用线程池替代直接Thread创建
//    private ExecutorService processorExecutor;
//
//    @ServiceActivator(inputChannel = "topic-commanager")
//    @Override
//    public void handleMessage(Message<?> message) throws MessagingException {
//        try {
//            // 队列满时阻塞等待，确保不丢弃数据
//            messageBuffer.put(message);
//        } catch (InterruptedException e) {
//            Thread.currentThread().interrupt();
//            log.error("消息入队被中断", e);
//        }
//
//    }
//    // 在类中添加以下方法
//    @javax.annotation.PostConstruct
//    public void initMessageProcessor() {
//        // 启动消息处理线程
//        processorExecutor = Executors.newFixedThreadPool(1, r -> {
//            Thread t = new Thread(r, "MqttMessageProcessor");
//            t.setDaemon(true);
//            return t;
//        });
//
//        processorExecutor.submit(this::processMessageQueue);
//    }
//
//    private void processMessageQueue() {
//        while (!Thread.currentThread().isInterrupted()) {
//            try {
//                Message<?> message = messageBuffer.take();
//                // 异步处理具体业务逻辑
//                processMessageContent(message);
//            } catch (InterruptedException e) {
//                Thread.currentThread().interrupt();
//                break;
//            } catch (Exception e) {
//                log.error("消息队列处理异常", e);
//            }
//        }
//    }
//
//    private void processMessageContent(Message<?> message) {
//        log.info("设备信息信息:{}",message);
//        String payload = (String) message.getPayload();
//        MessageHeaders headers = message.getHeaders();
//
//        String mqttReceivedTopic = headers.get("mqtt_receivedTopic", String.class);
//
//        //先解析出重要信息 根据topic分别进行消息处理
//        // 解析主题和消息内容（保持您原有的解析逻辑）
//        String[] topicParts = mqttReceivedTopic.split("/");
//        if (topicParts.length < 4) {
//            log.warn("主题格式错误: {}", mqttReceivedTopic);
//            return;
//        }
//        /*解析出订阅主题*/
//        String command = topicParts[3];
//        JSONObject jo = JSON.parseObject(payload);
//        JSONArray deviceArray = jo.getJSONArray("devices");
//
//        // 从设备数据中提取设备ID（station_no）
////        String stationNo = extractStationNo(deviceArray);
//        String stationNo = topicParts[4].split(":")[0];
//        switch (command){
//            case Common.Driver.Type.UP_BIZ: {handleUpBiz(stationNo,deviceArray); break;}
//            case Common.Driver.Type.HEART_BEAT: { handleHeartbeat(stationNo); break;}
//            default:{
//                log.info("暂不支持的主题操作:{}",command);
//            }
//        }
//    }
//
//    /**
//     * 定时检查（新增，作为消息触发的补充）
//     */
//    @Scheduled(fixedRate = 10000)
//    public void scheduledCheck() {
//        checkAndSwitch();
//    }
//    /**
//     * 处理心跳消息（新增）
//     */
//    private void handleHeartbeat(String stationNo) {
//        // 更新设备最后心跳时间
//        lastHeartbeatTimes.put(stationNo, System.currentTimeMillis());
//        log.debug("收到设备[{}]的心跳", stationNo);
//
//        // 检查是否需要切换
//        checkAndSwitch();
//    }
//    /**
//     * 检查并执行切换（新增）
//     */
//    private void checkAndSwitch() {
//        if (isSwitching.get() || !shouldCheckSwitch()) {
//            return;
//        }
//
//        try {
//            isSwitching.set(true);
//
//            // 检查当前活跃设备是否在线
//            if (!isStationAlive(activeStationNo)) {
//                // 查找其他可用设备
//                for (String stationNo : lastHeartbeatTimes.keySet()) {
//                    if (!stationNo.equals(activeStationNo) && isStationAlive(stationNo)) {
//                        switchToStation(stationNo);
//                        break;
//                    }
//                }
//            }
//        } finally {
//            isSwitching.set(false);
//        }
//    }
//
//    private void switchToStation(String newStationNo) {
//        log.warn("主备切换: 从[{}]切换到[{}]", activeStationNo, newStationNo);
//        activeStationNo = newStationNo;
//        lastSwitchTime = System.currentTimeMillis();
//
//        // 这里可以添加切换后的回调逻辑
//        onStationSwitched(newStationNo);
//    }
//    /**
//     * 设备切换回调（可扩展）
//     */
//    private void onStationSwitched(String newStationNo) {
//        // 示例：记录切换日志或通知其他服务
//        log.info("已切换到设备: {}", newStationNo);
//    }
//
//    /**
//     * 检查设备是否存活（新增）
//     */
//    private boolean isStationAlive(String activeStationNo) {
//        Long lastHeartbeat = lastHeartbeatTimes.get(activeStationNo);
//        return lastHeartbeat != null &&
//                (System.currentTimeMillis() - lastHeartbeat) < HEARTBEAT_TIMEOUT_MS;
//    }
//
//    /**
//     * 判断是否应该检查切换（新增）
//     */
//    private boolean shouldCheckSwitch() {
//        return System.currentTimeMillis() - lastSwitchTime > SWITCH_COOLDOWN_MS;
//    }
//
//    /**
//     * 从设备数据中提取站点编号（新增）
//     */
//    private String extractStationNo(JSONArray deviceArray) {
//        try {
//            JSONObject device = deviceArray.getJSONObject(0);
//            JSONArray services = device.getJSONArray("services");
//            JSONObject service = services.getJSONObject(0);
//            JSONObject data = service.getJSONObject("data");
//            return data.getString("station_no");
//        } catch (Exception e) {
//            log.warn("提取设备ID失败: {}", e.getMessage());
//            return null;
//        }
//    }
//
//    private void handleUpBiz(String stationNo , JSONArray payload){
//        // 只处理当前活跃设备的业务消息
//        /*判断stationNo是否有：*/
//        if (stationNo.contains(":")){
//            stationNo = stationNo.split(":")[0];
//        }
//        if (stationNo.equals(activeStationNo)) {
//            log.info("处理活跃设备[{}]的业务消息: {}", stationNo, payload);
//            upBizService.doUpBiz(payload);
//        } else {
//            log.debug("忽略非活跃设备[{}]的业务消息", stationNo);
//        }
//    }
//}
